package broker

import (
	"context"
	"encoding/json"
	"errors"
	"github.com/Shopify/sarama"
	"github.com/google/uuid"
	"log"
	"sync"
)

type kBroker struct {
	addrs     [] string
	c         sarama.Client
	p         sarama.SyncProducer
	sc        []sarama.Client
	connected bool
	scMutex   sync.RWMutex
	opts      Options
}

type subscriber struct {
	cg   sarama.ConsumerGroup
	t    string
	opts SubscribeOptions
}

type publication struct {
	t string
	err error
	cg sarama.ConsumerGroup
	km *sarama.ConsumerMessage
	m *Message
	sess sarama.ConsumerGroupSession
}

func (p *publication) Topic() string{
	return p.t
}

func (p *publication) Message() *Message {
	return p.m
}

func (p *publication) Value() []byte{
	return p.m.Value
}

func (p *publication) Ack() error{
	p.sess.MarkMessage(p.km, "")
	return nil
}

func (p *publication) Error() error{
	return p.err
}

func (s *subscriber) Options() SubscribeOptions {
	return s.opts
}

func (s *subscriber) Topic() string{
	return s.t
}

func (s *subscriber) Unsubscribe() error{
	return s.cg.Close()
}

func (k *kBroker) Address() string{
	if len(k.addrs)>0{
		return k.addrs[0]
	}
	return "127.0.0.1:9092"
}

func (k *kBroker) Connect() error{
	if k.isConnected(){
		return nil
	}
	k.scMutex.Lock()
	if k.c !=nil{
		k.connected=true
		k.scMutex.Unlock()
		return nil
	}
	k.scMutex.Unlock()
	pconfig :=k.getBrokerConfig()
	pconfig.Producer.Return.Successes = true
	pconfig.Producer.Return.Errors = true

	c, err := sarama.NewClient(k.addrs,pconfig)
	if err !=nil{
		return nil
	}

	p, err := sarama.NewSyncProducerFromClient(c)

	k.scMutex.Lock()
	k.c =c
	k.p =p
	k.sc = make([]sarama.Client,0)
	k.connected = true
	k.scMutex.Unlock()
	return nil
}

func (k *kBroker) Disconnect() error{
	if !k.isConnected(){
		return nil
	}
	k.scMutex.Lock()
	defer k.scMutex.Unlock()
	for _,client :=range k.sc{
		client.Close()
	}
	k.sc=nil
	k.p.Close()
	if err := k.c.Close(); err !=nil{
		return err
	}
	k.connected=false
	return nil
}

func (k *kBroker) Options() Options {
	return k.opts
}

func (k *kBroker) Publish(topic string, msg *Message, opts ...PublishOption) error{
	if !k.isConnected(){
		return errors.New("[kafka] broker not connected")
	}
	// 设置一些属性
	publishOptions := PublishOptions{}
	for _,o:= range opts{
		o(&publishOptions)
	}
	b,err :=json.Marshal(msg)
	if err !=nil{
		return nil
	}
	pMsg :=&sarama.ProducerMessage{
		Topic: topic,
		Value: sarama.ByteEncoder(b),
	}
	if publishOptions.Partitions !=0{
		pMsg.Partition =publishOptions.Partitions
	}
	if publishOptions.Key != ""{
		pMsg.Key = sarama.ByteEncoder(publishOptions.Key)
	}

	_,_,err =k.p.SendMessage(pMsg)
	return err
}


func (k *kBroker) isConnected() bool{
	k.scMutex.RLock()
	defer k.scMutex.RUnlock()
	return k.connected
}

func (k *kBroker) String() string{
	return "kafka"
}

func (k *kBroker) Init(opts ...Option) error{
	for _,o:= range opts{
		o(&k.opts)
	}
	var cAddrs []string
	for _,addr := range k.opts.Addrs{
		if len(addr) ==0{
			continue
		}
		cAddrs = append(cAddrs,addr)
	}
	if len(cAddrs)==0{
		cAddrs = []string{"127.0.0.1:9092"}
	}
	k.addrs = cAddrs
	return nil
}


func (k *kBroker) getSaramaClusterClient(topic string) (sarama.Client, error) {
	config := k.getClusterConfig()
	cs, err := sarama.NewClient(k.addrs, config)
	if err != nil {
		return nil, err
	}
	k.scMutex.Lock()
	defer k.scMutex.Unlock()
	k.sc = append(k.sc, cs)
	return cs, nil
}

func (k *kBroker) Subscribe(topic string, handler Handler, opts ...SubscribeOption) (Subscriber, error){
	opt := SubscribeOptions{
		AutoAck: true,
		Queue: uuid.New().String(),
	}
	for _,o := range opts{
		o(&opt)
	}

	c, err := k.getSaramaClusterClient(topic)
	if err !=nil{
		return nil,err
	}

	cg, err := sarama.NewConsumerGroupFromClient(opt.Queue,c)
	if err !=nil{
		return nil,err
	}
	h := &consumerGroupHandler{
		handler: handler,
		subopts: opt,
		kopts: k.opts,
		cg:cg,
	}
	ctx := context.Background()
	topics := []string{topic}
	go func(){
		for{
			select{
			case err:= <-cg.Errors():
				if err !=nil{
					log.Println("consumer error:",err)
				}
			default:
				err := cg.Consume(ctx,topics,h)
				switch err{
				case sarama.ErrClosedConsumerGroup:
					return
				case nil:
					continue
				default:
					log.Println(err)
				}
			}
		}
	}()

	return &subscriber{cg:cg, opts:opt,t:topic},nil
}

func NewBroker(opts ...Option) Broker {
	options := Options{
		Context: context.Background(),
	}
	for _,o := range opts{
		o(&options)
	}
	var clusterAddrs []string
	for _,addr := range options.Addrs{
		if len(addr)==0{
			continue
		}
		clusterAddrs = append(clusterAddrs,addr)
	}
	return &kBroker{
		addrs: clusterAddrs,
		opts: options,
	}

}

func (k *kBroker) getBrokerConfig() *sarama.Config{
	if c,ok:=k.opts.Context.Value(brokerConfigKey{}).(*sarama.Config); ok{
		return c
	}
	return DefaultBrokerConfig
}

func (k *kBroker) getClusterConfig() *sarama.Config{
	if c,ok:=k.opts.Context.Value(clusterConfigKey{}).(*sarama.Config); ok{
		return c
	}
	clusterConfig := DefaultClusterConfig
	if !clusterConfig.Version.IsAtLeast(sarama.V0_10_2_0){
		clusterConfig.Version = sarama.V0_10_2_0
	}
	clusterConfig.Consumer.Return.Errors = true
	clusterConfig.Consumer.Offsets.Initial = sarama.OffsetNewest
	return clusterConfig
}

